package com.shujia.tour

import com.shujia.util.{Geography, SparkTool}
import org.apache.spark.sql.expressions.{UserDefinedFunction, Window}
import org.apache.spark.sql.{DataFrame, SparkSession}

object DalTourProvinceTouristApp extends SparkTool {
  /**
    * 在子类中实现run方法，实现自定义的代码逻辑
    *
    * 在子类run方法的前面加上下面两行代码
    * import spark.implicits._
    * import org.apache.spark.sql.functions._
    *
    * @param spark spark的环境
    */
  override def run(spark: SparkSession): Unit = {
    import org.apache.spark.sql.functions._
    import spark.implicits._

    //停留表
    val stayPoint: DataFrame = spark.sql(s"select * from dwd.dwd_staypoint_msk_d where day_id=$day_id")

    //用户画像表
    val usertag: DataFrame = spark.sql(s"select * from dim.dim_usertag_msk_m where month_id=$month_id")

    //行政区配置表
    val adminCode: DataFrame = spark.sql("select * from  dim.dim_admincode")


    /**
      * 游客过滤逻辑
      * 1、停留时间大于3小时
      * 2、出游距离大于10KM  ，省内停留点到常住地的最远距离
      *
      */

    val compute_distance: UserDefinedFunction = udf((grid1: String, grid2: String) => {
      //计算两个网格点的距离
      //返回单位市km
      val distance: Double = Geography.calculateLength(grid1.toLong, grid2.toLong) / 1000
      distance
    })


    //2、出游距离大于10KM
    val resultDF: DataFrame = stayPoint
      .join(usertag, "mdn")
      //计算每个点到常住地的距离
      .withColumn("distance", compute_distance($"resi_grid_id", $"grid_id"))
      //关联行政区配置表
      .join(adminCode.hint("broadcast"), "county_id")
      //获取每个人在市内距离常住地最远的距离
      .withColumn("d_max_distance", max($"distance") over Window.partitionBy($"mdn", $"prov_id"))
      //出游距离大于10KM
      .where($"d_max_distance" > 10)

      //计算一个人在一个市内的停留时间
      .groupBy($"mdn", $"prov_id", $"d_max_distance", $"resi_county_id")
      .agg(sum($"duration") as "d_stay_time")
      //停留时间大于3小时
      .where($"d_stay_time" > 180)

      //整理数据
      .select($"mdn", $"resi_county_id" as "source_county_id", $"prov_id" as "d_province_id", round($"d_stay_time", 2), round($"d_max_distance", 3))

      //合并小文件
      .coalesce(2)

    //保存数据
    saveDataAndAddPartitionWithDay(
      resultDF,
      "/daas/motl/dal_tour/dal_tour_province_tourist_msk_d",
      "dal_tour.dal_tour_province_tourist_msk_d"
    )

  }
}
